package cn.liuxb.spa;

import java.io.IOException;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;

import cn.liuxb.spa.OrderServiceMock.PayParam;
import cn.liuxb.spa.OrderServiceMock.PayResult;
import cn.liuxb.spa.OrderServiceMock.TradeStatus;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.util.CharsetUtil;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;

/**
 * 应保证请求一定会被应答.
 * 
 * @author liuxianbo
 */
@Slf4j
@Data
@Component
public class SpaHandler {
	OrderServiceMock orderServiceMock = new OrderServiceMock().setSpaHandler(this);
	Map<String, SpaContext> requestCache = new ConcurrentHashMap<>();
	ScheduledExecutorService timeoutExecutor = Executors.newScheduledThreadPool(3);
	int timeOutSeconds = 5;
	ObjectMapper om = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
			.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
//	Map<String, BizHandler<?, ?>> uri2BizHandlerMap = new LinkedHashMap<>();

	@PostConstruct
	public void init() {
//		uri2BizHandlerMap.put("/pay", new PayHandler());

		// 超时自动应答处理中
		timeoutExecutor.scheduleAtFixedRate(() -> {
			List<String> timeoutKeys = requestCache.entrySet().stream()
					.filter(e -> e.getValue().reqTimeMs + timeOutSeconds * 1000 < System.currentTimeMillis())
					.map(Entry::getKey).collect(Collectors.toList());

			timeoutKeys.stream().map(requestCache::remove).filter(Objects::nonNull).forEach(spaCtx -> {
				spaCtx.respObj(new PayResult().setRetCode("9999").setRetMsg("后端业务服务超时!"));
				log.info("请求{}已超时,返回系统异常!", spaCtx.id());
				sendResponse(spaCtx);
			});

		}, 3, 1, TimeUnit.SECONDS);
	}

	// 收到请求
	public void onRequest(SpaContext sc) {
		log.info("请求报文:" + sc.reqStr);
		PayParam payParam = null;
		try {
			payParam = om.readValue(sc.reqStr, PayParam.class);
			sc.id(payParam.getOutTradeNo());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
		PayResult payResult = orderServiceMock.pay(payParam);
		log.info("收到同步应答:" + payResult);
		if (payResult.getTradeStatus() != TradeStatus.PROC) {
			sc.respObj(payResult);
			sendResponse(sc);
		} else {
			log.info("同步未返回明确结果,开始等待异步通知...");
			requestCache.put(payParam.getOutTradeNo(), sc);
		}
	}

	// 收到异步通知
	public void onAsyncNotify(PayResult payResult) {
		log.info("收到异步通知:" + payResult);
		SpaContext sc = requestCache.remove(payResult.getOutTradeNo());
		if (sc == null) {
			log.info("没有等待中的请求.");
			return;
		}
		sc.respObj(payResult);
		sendResponse(sc);
	}

	private void sendResponse(SpaContext sc) {
		try {
			sc.respStr(om.writeValueAsString(sc.respObj()));
		} catch (JsonProcessingException e) {
			throw new RuntimeException(e);
		}
		log.info("发送应答:" + sc.respStr());
		FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
				Unpooled.copiedBuffer(sc.respStr, CharsetUtil.UTF_8));
		response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=UTF-8");
		sc.ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
	}

	@Data
	@Accessors(fluent = true)
	public static class SpaContext {
		// 业务id,用于关联请求和异步通知
		String id;
		ChannelHandlerContext ctx;
		FullHttpRequest httpRequest;
		long reqTimeMs;
		String reqStr;
		Object reqObj;
		String respStr;
		Object respObj;
		Exception e;
	}

}
